package com.kgcx.common.handler;

import com.kgcx.common.rabbitmq.bean.MqFail;
import com.kgcx.common.rabbitmq.producer.MqFailProducer;
import com.kgcx.common.utils.CollectionUtils;
import com.kgcx.common.utils.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.RabbitListenerErrorHandler;
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Set;

/**
 * @author 吴刚林
 * @date 2017/11/29
 */
@Component
@SuppressWarnings("unchecked")
public class CustRabbitListenerErrorHandler implements RabbitListenerErrorHandler
{
    private static final Logger LOG = LoggerFactory.getLogger(CustRabbitListenerErrorHandler.class);

    private final String CONSUMEFAIL_MESSAGEID = "MQ_consumefail_messageId";

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private MqFailProducer mqFailProducer;

    @Scheduled(cron = "0 0 0 * * ?")   //每天0点触发
    public void cleanMessageIdRedis() {
        HashOperations hashOperations = stringRedisTemplate.opsForHash();
        Set keys = hashOperations.keys(CONSUMEFAIL_MESSAGEID);
        if (CollectionUtils.isNotEmpty(keys)) {
            keys.forEach(i -> hashOperations.delete(CONSUMEFAIL_MESSAGEID, i));
        }
    }

    @Override
    public Object handleError(Message amqpMessage, org.springframework.messaging.Message<?> message, ListenerExecutionFailedException exception) {
        String messageId = amqpMessage.getMessageProperties().getMessageId();
        String receivedExchange = amqpMessage.getMessageProperties().getReceivedExchange();
        String receivedRoutingKey = amqpMessage.getMessageProperties().getReceivedRoutingKey();
        String key = receivedExchange + "." + receivedRoutingKey + "." + messageId;
        LOG.error("messageId=" + key);

        HashOperations<String, Object, Object> hashOperations = stringRedisTemplate.opsForHash();
        Object o = hashOperations.get(CONSUMEFAIL_MESSAGEID, key);
        Integer count = 0;
        if (null != o) {
            count = Integer.valueOf(o.toString());
            LOG.info("count=" + count + ", messageId=" + key);
            if (count >= 10) {
                MqFail mqFail = new MqFail();
                mqFail.setRaw(new String(amqpMessage.getBody()));
                mqFail.setMessage(message.getPayload().toString());
                mqFail.setException(ExceptionUtils.getStackTrace(exception));
                mqFailProducer.mqFail(mqFail);
                hashOperations.delete(CONSUMEFAIL_MESSAGEID, key);
                return null;
            }
        }
        hashOperations.put(CONSUMEFAIL_MESSAGEID, key, (count + 1) + "");
        throw new AmqpException(exception);
    }
}
